package awk.disruptor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import java.util.concurrent.ThreadFactory;

/**
 * @description:
 * @author: wanghaifeng
 * @date: 2020/12/23
 *
 * disrupt工具类
 */
public final class DisruptorUtil {
    private static DisruptorUtil disruptorUtil;
    private final Disruptor<MessageEvent> disruptor;
    // RingBuffer 大小，必须是 2 的 N 次方；
    private static final int RING_BUFFER_SIZE = 1024 * 1024;
    public MessageEventProducer messageEventProducer;

    private DisruptorUtil() {
        //重点1
        disruptor = new Disruptor<>(MessageEvent::new, RING_BUFFER_SIZE, (ThreadFactory) Thread::new, ProducerType.SINGLE, new YieldingWaitStrategy());
        //重点2
        disruptor.handleEventsWith((event, l, b) -> System.out.println("消费者消费消息：" + event.getMessage()));
        disruptor.setDefaultExceptionHandler(new MessageExceptionHandler());
    }

    /**
     * 获取 LogDisruptorUtil 实例
     *
     * @return LogDisruptorUtil
     */
    public static DisruptorUtil getInstance() {
        if (disruptorUtil == null) {
            synchronized (DisruptorUtil.class) {
                if (disruptorUtil == null) {
                    disruptorUtil = new DisruptorUtil();
                    return disruptorUtil;
                }
            }
        }
        return disruptorUtil;
    }
    /**
     * 启动disruptor
     */
    public RingBuffer<MessageEvent> start() {
        RingBuffer<MessageEvent> ringBuffer = disruptor.start();
        messageEventProducer = new MessageEventProducer(ringBuffer);
        //应用关闭前关闭disrupt
        Runtime.getRuntime().addShutdownHook(new Thread(disruptor::shutdown));

        return ringBuffer;
    }

    /**
     * 生产者发布事件
     * 这个disruptorUtil持有的disruptor唯一实例调用messagemEventProduce生产数据，然后用其持有的 messageEventConsumer 取消费数据
     *
     * 如果此处 disruptor 持有一个生产者 messagemEventProduce ，同时有多个 messageEventConsumer ,那么不就是发布-订阅模型么
     *
     * 如果此处 disruptor 持有多个生产者 messagemEventProduce ，但只有一个 messageEventConsumer
     */
    public void produce(String message) {
        messageEventProducer.onData(message);
    }

}
